package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.analyze.client.WirelessStat._
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.{SQLContext, DataFrame}
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

/**
 * Description:推送转化统计清洗逻辑
 * Date:2017/3/14 18:03
 * Created by wxxuyuan
 */
object WirelessPushConversionRate {
  def process(sc: SparkContext, pageForward: DataFrame,date :String): Unit = {

    val commons = sc.broadcast(Seq("logType","deviceToken", "sessionId", "channelCode", "platformCode", "pageTypeCode",
      "pageParam", "buttonCode", "pageCode","holeCode","accessCode","timestamp"))

    //取出category_id
    val logWithCategoryId = pageForward.select(commons.value.map(col): _*)
      .withColumn("category_id", when(col("pageParam").getItem("categoryId").isNull, "-1")
        .otherwise(col("pageParam").getItem("categoryId").cast("string")))

    val ACCURATE_PUSH_PAGE_CODE : String = "3bnsc"               //精准推送页面
    val PAY_PAGE_CODE :String = "3DaiZ"                                    //支付页页面

    import com.lvmama.rhino.common.utils.JDBCUtil.Conversions._
    val iosCodePredicates = Array("name LIKE 'ios%'")
    val sqlContext = new SQLContext(sc)
    val iosCodeArr = sqlContext.loadFromMysql("statistic_continuity_code", iosCodePredicates)
      .coalesce(1)
      .select(col("code")).collect()//取出ios平台code码


    val pushLog = logWithCategoryId.filter(col("platformCode") =!= "2g53").coalesce(5)
      .withColumn("platform",when(col("platformCode").isin(iosCodeArr.map(_.get(0)): _*),"IPHONE")
        .otherwise("ANDROID"))
      .withColumn("product_id", when(col("pageParam").getItem("productId").isNull, "-1").otherwise(col("pageParam").getItem("productId")))
      .withColumn("pageTotalCode", concat(col("pageCode"), col("category_id"), col("product_id"), col("buttonCode")))
      .withColumn("pageTotalCodeLag", dataLag(col("pageTotalCode")))
      .withColumn("buttonLag", dataLag(col("buttonCode")))
      .filter(col("pageTotalCodeLag").isNull.or(col("pageTotalCode") !== col("pageTotalCodeLag"))).coalesce(5)
      .filter(col("buttonCode") !== "back")
      .filter(col("buttonLag").isNull.or(col("buttonLag") !== "back"))
      .drop(col("pageTotalCode"))
      .drop(col("pageTotalCodeLag"))
      .drop(col("buttonLag"))
      .drop(col("product_id"))
      .drop(col("platformCode"))
      .persist(StorageLevel.MEMORY_AND_DISK_SER)

    /**
      * 从MySQL 计算前一天的精准推送总量
      */
    val predicates = Array("push_category = 'ASSOCITAION' AND begin_time > '" + Utils.getYesterday() + "' AND begin_time < '"
      + Utils.getToday() + "'")
    val pushAmountGroup = sqlContext.loadFromMysqlHippo("push_message_statistic", predicates)
      .groupBy(col("platform"))
      .agg(sum("push_total_count").as("total_push_no"))

    /**
      * 从精准推送页日志计算点击量和uv
      */
    val pvUvGroup = pushLog.filter(col("pageTypeCode").equalTo(ACCURATE_PUSH_PAGE_CODE))
      .groupBy(col("platform"))
      .agg(count("*").as("click_no"),countDistinct("sessionId").as("uv"))

    /**
      * 从下单页计算订单相关指标   页面入口中第一级为推送页面code的都是
      */
    val orderGroup = pushLog.filter(col("pageTypeCode").equalTo(PAY_PAGE_CODE))
      .withColumn("entrance",split(col("accessCode"),"_")(0))
      .filter(col("entrance") === ACCURATE_PUSH_PAGE_CODE)
      .withColumn("orderId", when(col("pageParam").getItem("orderNum").isNull, "-1")
        .otherwise(col("pageParam").getItem("orderNum")))  //取出订单号
      .withColumn("orderAmount", when(col("pageParam").getItem("paymentAmount").isNull, 0)
      .otherwise(col("pageParam").getItem("paymentAmount")))  //取出订单金额
      .dropDuplicates(Seq("orderId"))
      .groupBy(col("platform"))
      .agg(countDistinct("sessionId").as("ordered_user_no"),
        count("orderId").as("order_total_no"),
        sum("orderAmount").as("order_total_amount"))

    val resultDf = pushAmountGroup.join(pvUvGroup,Seq("platform"),"outer").join(orderGroup,Seq("platform"),"outer")
      .na.fill(0,Seq("total_push_no","click_no","uv","ordered_user_no","order_total_no","order_total_amount"))

    //获取源数据的日期
    val dateOfSouceData = date.replaceAll("/","-")

    resultDf.withColumn("date",lit(dateOfSouceData))
      .withColumn("push_name",lit("ASSOCITAION"))
      .insertDF2MysqlDirect("accurate_push_stat")

    pushLog.unpersist()

  }

}
